package order;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * 顺序消息-生产者
 *
 * @author Administrator
 * @date 2022/03/31 15:20
 **/
@Slf4j
public class OrderProducer {
    public static void main(String[] args) throws MQBrokerException, RemotingException, InterruptedException, MQClientException {
        // 构建顺序生产者
        DefaultMQProducer producer = new DefaultMQProducer("Order_Producer");
        // NameServer
        producer.setNamesrvAddr("192.168.147.151:9876;192.168.147.152:9876");

        producer.start();
        List<Car> cars = Car.builderCar();
        for (Car car : cars) {
            // 消息
            Message message = new Message("Order-Topic", "order", car.toString().getBytes(StandardCharsets.UTF_8));

            // 消息队列选择 最后一个参数为 Object arg
            SendResult result = producer.send(message, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    long id = (long) arg;
                    // 根据 id 对队列长度取模，决定往哪个队列发
                    long queueNo = id % (mqs.size());

                    return mqs.get((int) queueNo);
                }
            }, car.getCarId());


            log.info("发送消息成功:{}",result);
            TimeUnit.MILLISECONDS.sleep(100);
        }
        producer.shutdown();
    }
}
